package org.huangrui.spark.java.streaming;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.Optional;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;

/**
 * nc -lp 9999
 * nc -L -p 9999
 * @Author hr
 * @Create 2024-10-21 20:22
 */
public class SparkStreaming05_State {
    public static void main(String[] args) throws InterruptedException {
        System.setProperty("HADOOP_USER_NAME", "huangrui");
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming");
        JavaStreamingContext jsc = new JavaStreamingContext(conf, new Duration(5*1000L));
        // 无状态数据操作，只对当前的采集周期内的数据进行处理
        // 在某些场合下，需要保留数据统计结果（状态），实现数据的汇总
        // 使用有状态操作时，需要设定检查点路径，将中间结果保存到文件系统中
        jsc.checkpoint("cp");
        JavaReceiverInputDStream<String> socketTextStream = jsc.socketTextStream("localhost", 9999);
        JavaPairDStream<String, Integer> mapToPair = socketTextStream.flatMap(line -> Arrays.asList(line.split(" ")).iterator()).mapToPair(word -> new Tuple2<>(word, 1));
        // JavaPairDStream<String, Integer> wordCountDS = mapToPair.reduceByKey(Integer::sum);
        // updateStateByKey：根据key对数据的状态进行更新
        // 传递的参数中含有两个值
        // 第一个值表示相同的key的value数据
        // 第二个值表示缓存区相同key的value数据
        mapToPair.updateStateByKey((List<Integer> values, Optional<Integer> state) -> {
            Integer val = values.stream().mapToInt(i -> i).sum();
            return state.isPresent() ? Optional.of(val + state.get()) : Optional.of(val);
        }).print();

        jsc.start();
        jsc.awaitTermination();
    }
}
